/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package cn.edu.thu.laud.objectstore.fieldmanager;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.Deletion;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.datanucleus.exceptions.NucleusDataStoreException;
import cn.edu.thu.laud.objectstore.CassandraConsistency;
import cn.edu.thu.laud.utils.serde.ConverterContext;
import cn.edu.thu.laud.objectstore.index.BatchMutation;

/**
 * Class that allows us to perform basic I/O ops on external entities
 * 
 * @author zhuoan
 * 
 */
public class ExternalEntityWriter extends ExternalEntity {

	private static final int ITERATION_SIZE = 100;

	private Set<ByteBuffer> savedColumns;

	public ExternalEntityWriter(Cassandra.Client client,
			ConverterContext context, String ownerColumnFamily,
			ByteBuffer rowKey, ByteBuffer ownerColumn) {
		super(client, context, ownerColumnFamily, rowKey, ownerColumn);
		savedColumns = new HashSet<ByteBuffer>();
	}

	/**
	 * Add the stored column to our internal queue
	 * 
	 * @param column
	 */
	protected void addStoredColumn(ByteBuffer column) {
		savedColumns.add(column);
	}

	/**
	 * Remove all columns from the collection/map. Useful for if a collection is
	 * set to null
	 */
	public void removeAllColumns(BatchMutation mutator) {

		byte[] columnBytes = ByteBufferUtil.getArray(ownerColumn);

		SliceRange range = new SliceRange();
		range.setCount(ITERATION_SIZE);
		range.setFinish(createBuffer(columnBytes, DELIM_MAX));

		SlicePredicate predicate = new SlicePredicate();
		predicate.setSlice_range(range);

		List<ColumnOrSuperColumn> results = null;

		do {

			range.setStart(createBuffer(columnBytes, DELIM_MIN));

			try {
				results = client.get_slice(rowKey, new ColumnParent(ownerColumnFamily),
						predicate, CassandraConsistency.get());
			} catch (Exception e) {
				// TODO Auto-generated catch block
				throw new NucleusDataStoreException(e.getMessage(), e);
			}
			List<ByteBuffer> toDelCols = new ArrayList<ByteBuffer>();
			// remove all columns that are presisted
			for (ColumnOrSuperColumn col : results) {
				toDelCols.add(col.getColumn().name);
			}
			
			if(toDelCols.size() >0){
				SlicePredicate pred = new SlicePredicate();
				pred.setColumn_names(toDelCols);				
				mutator.addDeletion(rowKey, Arrays.asList(new String[]{ownerColumnFamily}), new Deletion().setTimestamp(System
						.currentTimeMillis()).setPredicate(pred));
			}

			// advance our start key if required
			if (results.size() == ITERATION_SIZE) {
				columnBytes = results.get(ITERATION_SIZE - 1).getColumn().getName();
			}

		} while (results != null && results.size() == ITERATION_SIZE);

	}

	/**
	 * Removes all columns that have not been marked as persisted.
	 */
	public void removeRemaining(BatchMutation mutator) {
		
		byte[] columnBytes = ByteBufferUtil.getArray(ownerColumn);

		SliceRange range = new SliceRange();
		range.setCount(ITERATION_SIZE);
		range.setFinish(createBuffer(columnBytes, DELIM_MAX));

		SlicePredicate predicate = new SlicePredicate();
		predicate.setSlice_range(range);

		List<ColumnOrSuperColumn> results = null;

		do {

			range.setStart(createBuffer(columnBytes, DELIM_MIN));

			try {
				results = client.get_slice(rowKey, new ColumnParent(ownerColumnFamily),
						predicate, CassandraConsistency.get());
			} catch (Exception e) {
				// TODO Auto-generated catch block
				throw new NucleusDataStoreException(e.getMessage(), e);
			} 
			
			List<ByteBuffer> toDelCols = new ArrayList<ByteBuffer>();
			// remove all columns that are presisted
			for (ColumnOrSuperColumn col : results) {

				//not in our already saved columns, remove it
				if (!savedColumns.contains(col.getColumn().name)) {
					toDelCols.add(col.getColumn().name);
				}
			}
			
			if(toDelCols.size() >0){
				SlicePredicate pred = new SlicePredicate();
				pred.setColumn_names(toDelCols);				
				mutator.addDeletion(rowKey, Arrays.asList(new String[]{ownerColumnFamily}), new Deletion().setTimestamp(System
						.currentTimeMillis()).setPredicate(pred));
			}

			// advance our start key if required
			if (results.size() == ITERATION_SIZE) {
				columnBytes = results.get(ITERATION_SIZE - 1).getColumn().getName();
			}

		} while (results.size() == ITERATION_SIZE);

	}

	protected ByteBuffer createBuffer(byte[] columnBytes, byte delimByte) {
		
		ByteBuffer buffer = ByteBuffer.allocate(columnBytes.length + 1);
		buffer.mark();
		buffer.put(columnBytes);
		buffer.put(delimByte);
		buffer.reset();

		return buffer;
	}

}
